package com.github.smartheye.fabric.performance.spring.service.impl;

import static java.lang.String.format;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.FutureTask;

import org.hyperledger.fabric.protos.peer.FabricProposal.Proposal;
import org.hyperledger.fabric.protos.peer.FabricProposalResponse.ProposalResponse;
import org.hyperledger.fabric.sdk.GrpcPeerCallListener;
import org.hyperledger.fabric.sdk.HFClient;
import org.hyperledger.fabric.sdk.Peer;
import org.hyperledger.fabric.sdk.transaction.TransactionContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.protobuf.ByteString;

import io.grpc.Status;
import io.grpc.StatusRuntimeException;

public class GrpcPeerCallListenerImpl extends java.util.concurrent.CompletableFuture<List<org.hyperledger.fabric.sdk.ProposalResponse>> implements GrpcPeerCallListener {

	private Logger logger = LoggerFactory.getLogger(GrpcPeerCallListenerImpl.class);

	List<org.hyperledger.fabric.sdk.ProposalResponse> responseList = new ArrayList<org.hyperledger.fabric.sdk.ProposalResponse>();
	ConcurrentHashMap<ByteString, List<org.hyperledger.fabric.sdk.ProposalResponse>> responseMap = new ConcurrentHashMap<ByteString, List<org.hyperledger.fabric.sdk.ProposalResponse>>();
	HFClient hfclient;
	Collection<Peer> peers;
	TransactionContext transactionContext;
	Proposal proposal;

	@Override
	public void onMessage(Peer peer, ProposalResponse response) {

	}

	@Override
	public void onClose(Peer peer, Status status, ProposalResponse response, Exception e) {
		String peerName = peer.getName();
		String message = null;
		int statusCode = 500;
		if (status.isOk()) {
			message = response.getResponse().getMessage();
			statusCode = response.getResponse().getStatus();
		} else {
			Throwable cause = e.getCause();
			if (cause instanceof StatusRuntimeException) {
				message = format("Sending proposal to " + peerName + " failed because of: gRPC failure=%s", ((StatusRuntimeException) cause).getStatus());
			} else {
				message = format("Sending proposal to " + peerName + " failed because of: %s", cause.getMessage());
			}
			System.out.println(peerName+", message="+message);
			
			logger.error(message, new Exception(cause)); //wrapped in exception to get full stack trace.
		}
		org.hyperledger.fabric.sdk.ProposalResponse proposalResponse = new org.hyperledger.fabric.sdk.ProposalResponse(transactionContext.getTxID(),
		        transactionContext.getChannelID(), statusCode, message);
		proposalResponse.setProposalResponse(response);
		proposalResponse.setProposal(proposal);
		proposalResponse.setPeer(peer);

		responseList.add(proposalResponse);
		if(this.isDone()){
			return;
		}
		if (status.isOk()) {
			synchronized (responseMap) {
				if(this.isDone()){
					return;
				}
				ByteString payload = response.getPayload();
				List<org.hyperledger.fabric.sdk.ProposalResponse> resultList = responseMap.get(payload);
				if(resultList==null){
					resultList = new ArrayList<org.hyperledger.fabric.sdk.ProposalResponse>();
					responseMap.put(payload, resultList);
				}
				resultList.add(proposalResponse);
				
				if(resultList.size()>=2){
					//cancel all;
					this.complete(resultList);
				}
			}
		}
		
		if(responseList.size()==peers.size()){
			this.complete(responseList);
		}
	}

	@Override
	public void setHFClient(HFClient hfclient) {
		this.hfclient = hfclient;
	}

	@Override
	public void setTransactionContext(TransactionContext transactionContext) {
		this.transactionContext = transactionContext;
	}

	@Override
	public void setProposal(Proposal proposal) {
		this.proposal = proposal;
	}

	@Override
	public void setPeers(Collection<Peer> peers) {
		this.peers = peers;
	}

}
